package com.shujia.flink.table

import org.apache.flink.table.api.{EnvironmentSettings, Table, TableEnvironment}

object Demo2SQL {
  def main(args: Array[String]): Unit = {
    //创建环境设置对象
    val settings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .inStreamingMode() //流处理模式，输出连续结果
      //.inBatchMode()//批处理模式，输出最终结果
      .build()

    //创建flink sql环境
    val tEnv: TableEnvironment = TableEnvironment.create(settings)

    /**
     * 1、基于kafka的topic创建表 --- 动态表   -- source 表
     * executeSql: 可以执行DDl, DQL， DML
     */
    tEnv.executeSql(
      """
        |CREATE TABLE words (
        |   word STRING
        |) WITH (
        |  'connector' = 'kafka',-- 连接的数据源
        |  'topic' = 'words', -- topic
        |  'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092', -- broker列表
        |  'properties.group.id' = 'testGroup', -- 消费者组
        |  'scan.startup.mode' = 'earliest-offset', -- 读取数据的位置
        |  'format' = 'csv' --数据格式
        |)
        |""".stripMargin)


    /**
     * 2、创建一个sink表，用于结果查询结果
     */
    tEnv.executeSql(
      """
        |CREATE TABLE print_table (
        | word STRING,
        | c BIGINT
        |) WITH (
        | 'connector' = 'print'
        |)
        |""".stripMargin)

    /**
     * 3、基于动态表做连续查询, 返回一个新的动态表
     */
    tEnv.executeSql(
      """
        |insert into print_table
        |select word,count(1) as c
        |from
        |words
        |group by word
        |""".stripMargin)
  }
}
